package com.zyh.rcm.pvuv;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import com.zyh.rcm.entity.UniqueVisitorResult;
import com.zyh.rcm.entity.UserBehaviorEvent;
import org.apache.flink.api.common.state.ReducingStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.CountEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.streaming.api.windowing.triggers.PurgingTrigger;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.util.HashSet;
import java.util.Properties;
import java.util.Set;

public class UVJob1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置时间语义,默认是processTime
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//        DataStreamSource<String> dataStream = environment.socketTextStream("hadoop10", 8989);

        /*
            准备参数：
                kafka集群的ip和port
                topic的名字
                数据的反序列化方式
                消费者所属组的名字
         */
        String topicName = "topic-pv-uv";
//        SimpleStringSchema deserialization = new SimpleStringSchema();
        //自定义反序列化方式,从kafka拉取数据直接转换为实体对象
        UserBehaviorEventDeserialization deserialization = new UserBehaviorEventDeserialization();
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka24:9092");
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "flink");

        FlinkKafkaConsumer<UserBehaviorEvent> kafkaConsumer = new FlinkKafkaConsumer<>(topicName, deserialization, props);
        //在kafka消费端抽取水位线,越早抽取越好控制
        kafkaConsumer.assignTimestampsAndWatermarks(new UserBehaviorEventWaterMarkerAssigner(2000));

        DataStreamSource<UserBehaviorEvent> dataStream = environment.addSource(kafkaConsumer);

        SingleOutputStreamOperator<UniqueVisitorResult> result = dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(100)))
                //触发器
                .trigger(PurgingTrigger.of(ContinuousEventTimeTrigger.of(Time.seconds(5))))
                //剔除器
                //.evictor(CountEvictor.of(0,true))
                .process(new ProcessAllWindowFunction<UserBehaviorEvent, UniqueVisitorResult, TimeWindow>() {
                    private ValueState<BloomFilter<Integer>> bloomValueState;
                    private ValueState<Integer> sizeValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<BloomFilter<Integer>> filterValueStateDescriptor = new ValueStateDescriptor<>("bloomFilter", (Class<BloomFilter<Integer>>)(Class<?>)BloomFilter.class);
                        this.bloomValueState = getRuntimeContext().getState(filterValueStateDescriptor);
                        ValueStateDescriptor<Integer> size = new ValueStateDescriptor<>("size", Integer.class);
                        this.sizeValueState = getRuntimeContext().getState(size);
                    }

                    @Override
                    public void process(Context context, Iterable<UserBehaviorEvent> elements, Collector<UniqueVisitorResult> collector) throws Exception {
                        //对elements去重
//                        Set<Integer> set = new HashSet<>();
//                        for (UserBehaviorEvent element : elements) {
//                            set.add(element.getUserId());
//                        }
//                        int size = set.size();
                        BloomFilter<Integer> bloomFilter = this.bloomValueState.value();
                        if (bloomFilter==null){
                            bloomFilter = BloomFilter.create(Funnels.integerFunnel(), 10000, 0.0001);
                        }
                        //布隆过滤器
                        Integer size = this.sizeValueState.value();
                        if (size==null){
                            size = 0;
                        }
                        for (UserBehaviorEvent element : elements) {
                            if (!bloomFilter.mightContain(element.getUserId())){
                                size++;
                                bloomFilter.put(element.getUserId());
                            }
                        }
                        this.sizeValueState.update(size);
                        this.bloomValueState.update(bloomFilter);

                        TimeWindow window = context.window();
                        long start = window.getStart();
                        long end = window.getEnd();
                        collector.collect(new UniqueVisitorResult(start, end, size));
                    }
                });
        result.print();

        environment.execute("UVJob");
    }
}
